﻿#pragma once
//mqtt类,支持多个mqtt实例,
//支持绑定同一个或者不同的回调函数
//不同的mqtt实例体现的是不同的通信信道
#include "mqtt/mosquitto.h"
#include "export.h"
#include "share_data.h"
#include "msg_type.h"
#include <string>
#include <functional>
#include <map>
#include <vector>
#include "queue.h"
#define GET_MQTT_HANDLE MqttHandler::GetInstance()
class MqttClient;
//1.address 2.topic 3.payload 4.payloadlen
using mqtt_callback_t = std::function<void(const char*,const char*, const void*, int32_t)>;
//子mqtt客户端回调
using mqtt_callback_child_t = std::function<void(const char*, const void*, int32_t)>;
inline std::string make_address(std::string _ip, std::string _port)
{
	return _ip + ":" + _port;
}
class mqtt_ins_t
{
public:
	struct mosquitto* mosq{ nullptr };
	std::string note;
	std::string channel_name;
	int channel_id;
	std::string server_ip;
	int server_port;
	std::map<std::string, std::vector<mqtt_callback_child_t>> callback_map;
	std::map<std::string, std::vector<mqtt_callback_child_t>> special_callback_map;
	MqttClient* handler;
};
//封装单个mosquitto句柄，完成发布订阅内容
//绑定、数据回调在MqttHanlder中统一处理
class  DEMO_EXPORT MqttClient
{
public:
	MqttClient(const char* _ip, int _port,const char* _channel_name);
	void Start();
	void SubTopic(const char*);
	void SubTopic(const char*, mqtt_callback_child_t);
	void SubTopicWildcard(const char*, mqtt_callback_child_t);
	void Publish(const char*, const void*, int32_t _len, uint8_t qos = 0);
	void SetNote(const char* _note)
	{
		handler_.note = _note;
	}
	void SetChannelId(int _id)
	{
		handler_.channel_id = _id;
	}
	mqtt_ins_t& GetHandler()
	{
		return handler_;
	}	
	int64_t last_recv_stamp_{ -1 };
private:
	void default_callback(const char*,const void*,int32_t);
	mqtt_ins_t handler_;
};

class DEMO_EXPORT MqttHandler
{
public:
	static MqttHandler& GetInstance()
	{
		static MqttHandler handler;
		return handler;
	}
	void Start();
	static std::vector<std::string> SplitString(const std::string&,const std::string&);
	static void callback(struct mosquitto*, void*, const struct mosquitto_message*);
	static void onconnect(struct mosquitto*, void*, int);
	void send_sig(const char* _topic, const void* _pdata, int32_t _len, struct mosquitto* _mosq);
	void Add_Client(MqttClient* _client,const char* _channel)
	{
		map_.insert(std::make_pair(std::to_string((int64_t)_client->GetHandler().mosq),&_client->GetHandler()));
		ins_map_.insert(std::make_pair(_channel,_client->GetHandler().handler));
	}
	void Publish(const char* _topic,const std::string _payload,uint8_t qos=0)
	{
		using namespace msgType;
		auto com = GET_SHARE_DATA(pb_communicate);
		auto identify = Setting::GetInstance().GetIdentify();
		auto topic = (identify + "/" + std::string(_topic));
		std::string channel = "local";
		if (total_channel_.size() > com->current_channel_index())
		{
			channel = std::get<1>(total_channel_[com->current_channel_index()]);
			topic += "/" + channel;
		}
		std::string s = _payload;
		if ((*this)[channel.c_str()])
		{
			(*this)[channel.c_str()]->Publish(topic.c_str(),s.c_str(),s.length(),qos);
		}
	}
	void Publish(const char* _topic, const char* _payload,uint32_t _len, uint8_t qos = 0)
	{
		using namespace msgType;
		if ((*this)["net"])
		{
			(*this)["net"]->Publish(_topic, _payload, _len, qos);
		}
	}
	void PublishOrigin(const char* _topic, const google::protobuf::Message& _pdata)
	{
		using namespace msgType;
		auto com = GET_SHARE_DATA(pb_communicate);
		std::string channel = "local";
		if (total_channel_.size() > com->current_channel_index())
		{
			channel = std::get<1>(total_channel_[com->current_channel_index()]);
		}
		std::string s;
		_pdata.SerializeToString(&s);
		if ((*this)[channel.c_str()])
		{
			(*this)[channel.c_str()]->Publish(_topic, s.c_str(), s.length(), 0);
		}
	}
	void Publish(const char* _topic, const google::protobuf::Message& _pdata, uint8_t qos = 0)
	{
		using namespace msgType;
		auto com = GET_SHARE_DATA(pb_communicate);
		auto identify = Setting::GetInstance().GetIdentify();
		auto topic = (identify +"/" + std::string(_topic));
		std::string channel = "local";
		if (total_channel_.size() > com->current_channel_index())
		{
			channel = std::get<1>(total_channel_[com->current_channel_index()]);
			topic += "/" + channel;
		}
		std::string s;
		_pdata.SerializeToString(&s);
		if ((*this)[channel.c_str()])
		{
			(*this)[channel.c_str()]->Publish(topic.c_str(), s.c_str(), s.length(), qos);
		}
	}
	void Publish(const google::protobuf::Message& _pdata, uint8_t qos = 0)
	{
		using namespace msgType;
		auto com = GET_SHARE_DATA(pb_communicate);
		auto identify = Setting::GetInstance().GetIdentify();
		auto topic = identify + "/ctrl/" + _pdata.GetDescriptor()->name();
		std::string channel = "local";
		if (total_channel_.size() > com->current_channel_index())
		{
			channel = std::get<1>(total_channel_[com->current_channel_index()]);
			//if (channel == "tt") channel = "radio";
			topic += "/" + channel;
		}
		std::string s;
		_pdata.SerializeToString(&s);
		if ((*this)[channel.c_str()])
		{
			(*this)[channel.c_str()]->Publish(topic.c_str(), s.c_str(), s.length(), qos);
			if (mqtt_debug_)
			{
				topic = identify + "/debug/ctrl/" + _pdata.GetDescriptor()->name();
				topic += "/" + channel;
				(*this)[channel.c_str()]->Publish(topic.c_str(), _pdata.DebugString().c_str(),_pdata.DebugString().length());
			}
		}
	}
	void Sub(const char* _topic, mqtt_callback_child_t _func)
	{
		using namespace msgType;
		auto identify = Setting::GetInstance().GetIdentify();
		if ((*this)["local"])
		{
			(*this)["local"]->SubTopic((identify+"/" + _topic).c_str(), _func);
		}
	}
	void Sub(const char* _topic);
	MqttClient* operator[] (const char* _channel)
	{
		if (ins_map_.find(_channel) != ins_map_.end())
		{
			return ins_map_[_channel];
		}
		return nullptr;
	}
	std::vector<std::tuple<std::string,std::string>> GetTotalChannel();
private:
	MqttHandler();
	~MqttHandler();
	void pb_report2data(msgType::pb_report&,std::string);
	std::vector<std::tuple<std::string, std::string>> total_channel_;
	static std::map<std::string, mqtt_ins_t*> map_;
	static std::map<std::string, MqttClient*> ins_map_;
	static std::map<std::string, MqttClient*>  mqtt_map_;
	bool mqtt_debug_{ false };
};
